package com.pulse.base.mqtt;

import com.pulse.common.utils.spring.SpringUtils;
import com.pulse.base.service.ICtrlMsgService;
import com.pulse.base.service.IHeartbeatService;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * @ClassName MqttCallback
 * @Description
 * @Author 陆帅
 * @Date 2024/7/15 14:09
 */
@Component
public class MqttCallback implements MqttCallbackExtended {
    public final static Logger log = LoggerFactory.getLogger(MqttCallback.class);

    private MqttClient client;
    private MqttConnectOptions options;
    private String[] topic;
    private int[] qos;
    private String clientId;

    public MqttCallback(){}

    public MqttCallback(MqttClient client,MqttConnectOptions options, String[] topic, int[] qos, String clientId){
        this.client = client;
        this.options = options;
        this.topic = topic;
        this.qos = qos;
        this.clientId = clientId;
    }

    /**
     * mqtt连接后订阅主题
     * @param b
     * @param s
     */
    @Override
    public void connectComplete(boolean b, String s) {
        try {
            if(null != topic && null != qos) {
                if(client.isConnected()){
                    client.subscribe(topic,qos);
                    log.info("mqtt连接成功，客户端ID：" + clientId);
                    log.info("--订阅主题：" + Arrays.toString(topic));
                }else{
                    log.info("mqtt连接失败，客户端ID：" + clientId);
                }
            }
        }catch (Exception e) {
            log.info("mqtt订阅主题异常："+ e);
        }
    }

    /**
     * 断开重连
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.info("mqtt连接端口，发起重连......");
        try {
            if(null != client && !client.isConnected()) {
                client.reconnect();
                log.error("尝试重新连接");
            }else{
                client.connect(options);
                log.error("尝试建立新连接");
            }
        }catch (Exception e){
            log.error("异常。",e);
        }
    }

    /**
     * 消息处理 需要保存的数据有原始数据、系数、点位、topic、数据时间、接收时间
     * @param topic
     * @param mqttMessage
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        try {
            String[] topicSplit = topic.split("/");
            if(topicSplit.length!=4){
                return;
            }
            if("hrbt-up".equals(topicSplit[3])){ // 心跳
                IHeartbeatService heartbeatService = (IHeartbeatService) SpringUtils.getBean("HeartbeatService");
                if(heartbeatService != null){
                    heartbeatService.msgHandler(topic,mqttMessage);
                }
            }else if("ctrl-up".equals(topicSplit[3])){ // 数据
                ICtrlMsgService ctrlMsgService = (ICtrlMsgService) SpringUtils.getBean("CtrlMsgService");
                if(ctrlMsgService != null){
                    ctrlMsgService.msgHandler(topic,mqttMessage);
                }
            }
        }catch (Exception e) {
            log.error("异常。",e);
            log.info("处理mqtt消息异常："+ e + "\r" + e.getStackTrace() + e.getCause().getMessage());
        }
    }

    /**
     * 接收到消息调用令牌中调用
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }
}
